package com.link.serve.kafka.deprecated;

import java.util.*;
import java.util.concurrent.*;

import com.link.serve.constants.LinkConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.*;
import org.springframework.stereotype.Component;
import com.link.serve.kafka.proto.KafkaMessageProto.KafkaMessage;
import org.springframework.kafka.support.Acknowledgment;

@Slf4j
@Component
public class DynamicKafkaConsumerManager {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // 用于管理动态创建的容器，key 为容器ID
    private final Map<String, MessageListenerContainer> containerMap = new ConcurrentHashMap<>();

    /**
     * 动态创建并启动一个 Kafka 消费容器（批量消费模式），
     * 参数均动态传入，不依赖于 yml 中的配置。
     *
     * @param containerId      容器唯一ID
     * @param bootstrapServers Kafka集群地址，例如 "localhost:9092"
     * @param groupId          消费组ID
     * @param topic            要监听的 topic
     * @param partitions       要监听的分区数组；如果为 null 或长度为 0，则只指定 topic
     * @param keyDeserializer  Key的反序列化器，例如 StringDeserializer.class
     * @param valueDeserializer Value的反序列化器，例如你自定义的 KafkaMessageDeserializer.class
     */
    public void startContainer(String containerId,
                               String bootstrapServers,
                               String groupId,
                               String topic,
                               int[] partitions,
                               Class<?> keyDeserializer,
                               Class<?> valueDeserializer) {
        // 若已存在相同容器，先停止并移除
        if (containerMap.containsKey(containerId)) {
            stopContainer(containerId);
            containerMap.remove(containerId);
        }

        // 1. 构建 Consumer 配置（动态装配）
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 关闭自动提交
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
        // 如有需要可继续添加其他 Consumer 配置

        DefaultKafkaConsumerFactory<String, KafkaMessage> consumerFactory =
                new DefaultKafkaConsumerFactory<>(props);

        // 2. 构建 ContainerProperties
        ContainerProperties containerProps = new ContainerProperties(topic);// 只指定 topic

        // 设置 ACK 模式为 MANUAL（批量提交），注意：此处仅支持手动调用 acknowledgment.acknowledge()
        containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        // 可设置当 Topic 不存在时不抛出异常（如果需要）
        containerProps.setMissingTopicsFatal(false);

        // 3. 设置消息监听器，采用 BatchMessageListener 接口实现批量消费
        containerProps.setMessageListener(new BatchMessageListener<String, KafkaMessage>() {
            @Override
            public void onMessage(List<ConsumerRecord<String, KafkaMessage>> consumerRecords) {
            }
            public void onMessage(List<ConsumerRecord<String, KafkaMessage>> records, Acknowledgment acknowledgment) {
                try {
                    if (records == null || records.isEmpty()) {
                        acknowledgment.acknowledge();
                        return;
                    }
                    // 处理批量记录
                    batchProcessRecords(records);
                    // 手动提交 offset
                    acknowledgment.acknowledge();
                } catch (Exception e) {
                    log.error("[containerId={}] 批量处理消息时出现异常: {}", containerId, e.getMessage(), e);
                    // 出现异常时，可根据业务需求选择是否 ack
                }
            }
        });

        // 4. 创建 KafkaMessageListenerContainer（此类在 2.6.3 版本中可用）
        KafkaMessageListenerContainer<String, KafkaMessage> container =
                new KafkaMessageListenerContainer<>(consumerFactory, containerProps);

        // 5. 启动容器
        container.start();

        // 6. 保存到容器管理 Map 中
        containerMap.put(containerId, container);
        log.info("✅已启动 Kafka 容器 containerId=[{}], topic=[{}], partitions={}", containerId, topic, (partitions != null ? Arrays.toString(partitions) : "全部"));
    }

    /**
     * 停止指定的 Kafka 容器
     *
     * @param containerId 容器 ID
     */
    public void stopContainer(String containerId) {
        MessageListenerContainer container = containerMap.get(containerId);
        if (container != null && container.isRunning()) {
            container.stop();
            log.info("已停止 Kafka 容器 containerId=[{}]", containerId);
        } else {
            log.warn("停止容器失败：containerId=[{}] 不存在或已停止", containerId);
        }
    }

    /**
     * 批量处理消息记录，例如利用 Redis Pipeline 进行批量处理
     *
     * @param records 从 Kafka poll 到的消息记录列表
     */
    private void batchProcessRecords(List<ConsumerRecord<String, KafkaMessage>> records) {
        List<KafkaMessage> messageList = new ArrayList<>();
        for (ConsumerRecord<String, KafkaMessage> record : records) {
            KafkaMessage msg = record.value();
            if (msg != null) messageList.add(msg);
        }
        if (messageList.isEmpty()) return;

        // 使用 Redis Template 执行 pipeline 操作
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            // 对于每条消息，根据 operationType 进行处理
            for (KafkaMessage message : messageList) {
                String key = message.getKey();
                long serverId = message.getServerId();
                String operationType = message.getOperationType();
                if ("LOGIN".equals(operationType)) {
                    connection.stringCommands().setEx(
                            redisTemplate.getStringSerializer().serialize(key),
                            LinkConstant.REDIS_ONLINE_TIMEOUT_SECOND,
                            redisTemplate.getStringSerializer().serialize(String.valueOf(serverId))
                    );
                } else if ("HEARTBEAT".equals(operationType)) {
                    connection.keyCommands().expire(
                            redisTemplate.getStringSerializer().serialize(key),
                            LinkConstant.REDIS_ONLINE_TIMEOUT_SECOND
                    );
                } else if ("LOGOUT".equals(operationType)) {
                    // 根据实际需要处理 LOGOUT 操作，例如删除 key
                    // connection.keyCommands().del(...);
                } else {
                    log.warn("未知操作类型：{}", operationType);
                }
            }return null;
        });
        log.info("容器批量处理消息数量: {}", messageList.size());
    }
}